package cn.ac.iie.di.ban.ycl.cleaner;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;

public class MQReceiver {

    static final String mqNameServer = "10.136.64.29:9877";
//    static final String mqNameServer = "192.168.62.200:9876";
//    static final String mqNameServer = "192.168.11.94:9879";
    static final String topic = "swb_msg_1_mq";
//    static final String topic = "twb_msg_1_mq";
    static final String groupName = "zmc";

    public static void main(String[] args) throws REConnectionException {
        REConnection conn = new REConnection(mqNameServer);
        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(topic);
        builder.setGroupName(groupName);
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
        builder.setHandler(new REAbstractReceiveMessageHandler<REMessage>() {
            @Override
            public boolean handle(REMessage message) {
                //column name
                System.out.println("=== === === === === === === === === ===");
                StringBuilder sb = new StringBuilder();
                System.out.println("column name :");
                sb.append("{");
                for (int i = 0; i < message.colNames.length; i++) {
                    sb.append(message.colNames[i]).append(":").append(message.types[i]);
                    sb.append(",");
                }
                sb.append("}");
                System.out.println(sb.toString());

                REFieldType[] types = message.types;
                System.out.println("data :");
                int colLen = message.colNames.length;
                StringBuilder sb1 = new StringBuilder();
                for (Object[] data : message.data) {
                    sb1.append("{");
                    for (int i = 0; i < colLen; ++i) {
                        if (data[i] == null) {
                            continue;
                        }
                        sb1.append(message.colNames[i]).append(":").append(String.valueOf(data[i]));
                        sb1.append("  ");
                    }
                    sb1.append("}");
                    sb1.append("\n");
                    System.out.println(sb1.toString());
                }
                return true;
            }
        });
        builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
            @Override
            public boolean handle(byte[] message) {
                System.out.println("formatter fuilure ! size = " + message.length);
                return true;
            }
        });
        REReceiveSession session = (REReceiveSession) builder.build();
        session.start();
    }
}
